package com.hoe.tsb.fraft.client.role;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.hoe.tsb.Configuration;
import com.hoe.tsb.fraft.client.ClientRoleBridge;
import com.hoe.tsb.fraft.enums.RoleType;
import com.hoe.tsb.fraft.event.msg.FRaftMessage;
import com.hoe.tsb.fraft.exception.FRaftException;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Executor;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * <p>
 * 角色经纪人
 * </p>
 * 根据当前节点的角色类型选择合适的角色来进行消息处理
 *
 * @author hoe
 * @version 1.0
 * @date 2022/12/5 14:06
 */
@Slf4j
public class RoleAgent implements Role, RoleObserver {

    /**
     * 配置类
     */
    protected final Configuration mConfig;
    /**
     * 任期（节点刚启动时，默认任期是0，随后每轮选举会自增）
     */
    protected AtomicInteger term = new AtomicInteger(0);
    /**
     * 最大的溯源数据索引号
     * TODO 策略是”尽可能提交”，故该字段实际上没用
     */
    protected volatile Integer maxTraceIndex = 0;
    /**
     * 初始角色是跟随者
     */
    protected RoleType mRoleType = RoleType.FOLLOWER;
    /**
     * 创建一个3核心线程的线程池，1个用来发心跳，1个用来接收消息，1个用来取消息，其他的用来响应消息<br/>
     * 瓶颈在msgQueue中，这里8个线程足以应对
     */
    private Executor mExec = new ThreadPoolExecutor(3, 3,
            60L, TimeUnit.SECONDS, new SynchronousQueue<>(),
            new ThreadFactoryBuilder().setNameFormat("role-task-%d").build(),
            (Runnable r, ThreadPoolExecutor executor) -> { throw new FRaftException("已无空闲ClientWorker线程，请稍后再试"); });
    /**
     * 各角色的实现类
     */
    private final Role follower, candidate, leader;

    public RoleAgent(ClientRoleBridge bridge, Configuration configuration) {
        mConfig = configuration;
        follower = new Follower(bridge, configuration, this);
        candidate = new Candidate(bridge, configuration, this);
        leader = new Leader(bridge, configuration, this);

        //启动超时检测线程
        //Follower：Leader消息超时，切换为candidate；reset：收到任一leader消息
        //Candidate：投票数获取超时，回退回follower；reset：无
        //Leader：广播消息超时，发送心跳；reset：向成员广播了任一消息
        //该线程就是每x秒固定向角色进行timeout回调，由各角色自行判断回调是否生效，包括follower的随机超时时长也由角色内部进行处理
        mExec.execute(new Runnable() {
            @SuppressWarnings("all")
            @Override
            public void run() {
                while(true) {
                    synchronized (RoleAgent.class) {
                        try {
                            //Leader的超时时长比非Leader的少5s，防止因网络原因切换成Candidate
                            long waitMill = mRoleType == RoleType.LEADER ? mConfig.getTimeout() - 5000 : mConfig.getTimeout();
                            if (waitMill <= 25000) {
                                throw new FRaftException("配置错误，超时时长不能小于30秒");
                            }
                            RoleAgent.class.wait(waitMill);
                        } catch (Exception e) {
                            //ignore ...
                        }
                    }

                    try {
                        onTimeout();
                    } catch (Exception e) {
                        log.error("执行角色的超时逻辑错误", e);
                    }
                }
            }
        });
    }

    @Override
    public boolean handleMessage(FRaftMessage msg) {
        return getCurrent().handleMessage(msg);
    }

    @Override
    public void onTimeout() {
        getCurrent().onTimeout();
    }

    @Override
    public void onRoleTurn() {
        getCurrent().onRoleTurn();
    }

    @Override
    public void turnTo(RoleType roleType) {
        log.info("角色切换到：{}", roleType.name());
        mRoleType = roleType;
        onRoleTurn();
    }

    @Override
    public int getTerm() {
        return term.get();
    }

    @Override
    public void setTerm(int term) {
        log.info("任期切换到：{}", term);
        this.term.set(term);
    }

    @Override
    public int getMaxIndex() {
        return maxTraceIndex;
    }

    /**
     * 获取当前的角色类型美剧
     * @return 当前的角色类型
     */
    public RoleType getRoleType() {
        return mRoleType;
    }

    /**
     * 根据当前的角色枚举获取合适的角色
     * @return 当前的角色
     */
    private Role getCurrent() {
        switch (mRoleType) {
            case FOLLOWER:
                return follower;
            case CANDIDATE:
                return candidate;
            case LEADER:
                return leader;

        }

        throw new FRaftException("错误的角色类型");
    }
}
